package drds.server.execute_engine.data_handling;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.handler.MultiNodeQueryHandler;
import drds.server.memory.Memory;
import drds.server.memory.unsafe.memory.mm.DataNodeMemoryManager;
import drds.server.memory.unsafe.memory.mm.MemoryManager;
import drds.server.memory.unsafe.row.BufferHolder;
import drds.server.memory.unsafe.row.StructType;
import drds.server.memory.unsafe.row.UnsafeRow;
import drds.server.memory.unsafe.row.UnsafeRowWriter;
import drds.server.memory.unsafe.utils.PropertyConfig;
import drds.server.memory.unsafe.utils.sort.PrefixComparator;
import drds.server.memory.unsafe.utils.sort.PrefixComparators;
import drds.server.memory.unsafe.utils.sort.RowPrefixComputer;
import drds.server.memory.unsafe.utils.sort.UnsafeExternalRowSorter;
import drds.server.net.packet.ByteBufferUtil;
import drds.server.net.packet.BytesHolder;
import drds.server.net.packet.EofPacket;
import drds.server.net.packet.RowDataPacket;
import drds.server.route.Route;
import drds.server.server.SessionContext;

public class DataNodeMergeManager extends AbstractDataNodeMerge {

	private static Logger LOGGER = LoggerFactory.getLogger(DataNodeMergeManager.class);

	/**
	 * key为datanode的分片节点名字 value为对应的排序器 目前，没有使用！
	 */
	private ConcurrentHashMap<String, UnsafeExternalRowSorter> unsafeRows = new ConcurrentHashMap<String, UnsafeExternalRowSorter>();
	/**
	 * 全局sorter，排序器
	 */
	private UnsafeExternalRowSorter globalSorter = null;
	/**
	 * UnsafeRowGrouper
	 */
	private UnsafeRowGrouper unsafeRowGrouper = null;

	/**
	 * 全局merge，排序器
	 */
	private UnsafeExternalRowSorter globalMergeResult = null;

	/**
	 * sorter需要的上下文环境
	 */
	private final Memory Memory;
	private final MemoryManager memoryManager;
	private final PropertyConfig conf;
	/**
	 * Limit N，M
	 */
	private final int limitStart;
	private final int limitSize;

	public DataNodeMergeManager(MultiNodeQueryHandler handler, Route rrs) {
		super(handler, rrs);
		this.Memory = Server.getInstance().getMemory();
		this.memoryManager = Memory.getResultMergeMemoryManager();
		this.conf = Memory.getConf();
		this.limitStart = rrs.getLimitStart();
		this.limitSize = rrs.getLimitSize();
	}

	public void onRowMetaData(Map<String, Column> columToIndx, int fieldCount) throws IOException {

		OrderByColum[] orderCols = null;
		StructType schema = null;
		UnsafeExternalRowSorter.PrefixComputer prefixComputer = null;
		PrefixComparator prefixComparator = null;

		DataNodeMemoryManager dataNodeMemoryManager = null;
		UnsafeExternalRowSorter sorter = null;

		int[] groupColumnIndexs = null;
		this.fieldCount = fieldCount;

		if (route.getGroupByColumns() != null) {
			groupColumnIndexs = toColumnIndex(route.getGroupByColumns(), columToIndx);
			if (LOGGER.isDebugEnabled()) {
				for (int i = 0; i < route.getGroupByColumns().length; i++) {
					LOGGER.debug("groupColumnIndexs:" + route.getGroupByColumns()[i]);
				}
			}
		}

		if (route.getGroupByFilter() != null) {
			Column colMeta = columToIndx.get(route.getGroupByFilter().getLeft().toUpperCase());

			if (LOGGER.isDebugEnabled()) {
				LOGGER.debug("getHavingCols:" + route.getGroupByFilter().toString());
			}
			if (colMeta != null) {
				route.getGroupByFilter().setColumn(colMeta);
			}
		}

		if (route.isHasAggrColumn()) {
			List<AggregateFunctionType> mergCols = new LinkedList<AggregateFunctionType>();
			Map<String, Integer> mergeColsMap = route.getMergeColumnAndIndexMap();

			if (mergeColsMap != null) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("isHasAggrColumn:" + mergeColsMap.toString());
				}
				for (Map.Entry<String, Integer> mergEntry : mergeColsMap.entrySet()) {
					String colName = mergEntry.getKey().toUpperCase();
					int type = mergEntry.getValue();
					if (AggregateFunctionType.AVG == type) {
						Column sumColMeta = columToIndx.get(colName + "SUM");
						Column countColMeta = columToIndx.get(colName + "COUNT");
						if (sumColMeta != null && countColMeta != null) {
							Column colMeta = new Column(sumColMeta.columIndex, countColMeta.columIndex, sumColMeta.getColumType());
							mergCols.add(new AggregateFunctionType(colMeta, mergEntry.getValue()));
						}
					} else {
						Column colMeta = columToIndx.get(colName);
						mergCols.add(new AggregateFunctionType(colMeta, mergEntry.getValue()));
					}
				}
			}

			// add no alias merg column
			for (Map.Entry<String, Column> fieldEntry : columToIndx.entrySet()) {
				String colName = fieldEntry.getKey();
				int result = AggregateFunctionType.tryParseAggCol(colName);
				if (result != AggregateFunctionType.UNSUPPORT && result != AggregateFunctionType.NOMERGE) {
					mergCols.add(new AggregateFunctionType(fieldEntry.getValue(), result));
				}
			}

			/**
			 * Group操作
			 */
			unsafeRowGrouper = new UnsafeRowGrouper(columToIndx, route.getGroupByColumns(), mergCols.toArray(new AggregateFunctionType[mergCols.size()]), route.getGroupByFilter());
		}

		if (route.getOrderByColumnMap() != null) {
			LinkedHashMap<String, Integer> orders = route.getOrderByColumnMap();
			orderCols = new OrderByColum[orders.size()];
			int i = 0;
			for (Map.Entry<String, Integer> entry : orders.entrySet()) {
				String key = entry.getKey().toUpperCase();
				Column colMeta = columToIndx.get(key);
				if (colMeta == null) {
					throw new IllegalArgumentException("all columns in order by clause should be in the selected column list!" + entry.getKey());
				}
				orderCols[i++] = new OrderByColum(colMeta, entry.getValue());
			}

			/**
			 * 构造全局排序器
			 */
			schema = new StructType(columToIndx, fieldCount);
			schema.setOrderCols(orderCols);

			prefixComputer = new RowPrefixComputer(schema);

			if (orderCols.length > 0 && orderCols[0].getOrderType() == OrderByColum.ORDER_TYPE_ASC) {
				prefixComparator = PrefixComparators.LONG;
			} else {
				prefixComparator = PrefixComparators.LONG_DESC;
			}

			dataNodeMemoryManager = new DataNodeMemoryManager(memoryManager, Thread.currentThread().getId());

			/**
			 * 默认排序，只是将数据连续存储到内存中即可。
			 */
			globalSorter = new UnsafeExternalRowSorter(dataNodeMemoryManager, Memory, schema, prefixComparator, prefixComputer, conf.getSizeAsBytes(".buffer.pageSize", "1m"), false/**
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 是否使用基数排序
			 */
			, true/** 排序 */
			);
		}

		if (conf.getBoolean(".stream.output.result", false) && globalSorter == null && unsafeRowGrouper == null) {
			setStreamOutputResult(true);
		} else {

			/**
			 * 1.schema
			 */

			schema = new StructType(columToIndx, fieldCount);
			schema.setOrderCols(orderCols);

			/**
			 * 2 .PrefixComputer
			 */
			prefixComputer = new RowPrefixComputer(schema);

			/**
			 * 3 .PrefixComparator 默认是ASC，可以选择DESC
			 */

			prefixComparator = PrefixComparators.LONG;

			dataNodeMemoryManager = new DataNodeMemoryManager(memoryManager, Thread.currentThread().getId());

			globalMergeResult = new UnsafeExternalRowSorter(dataNodeMemoryManager, Memory, schema, prefixComparator, prefixComputer, conf.getSizeAsBytes(".buffer.pageSize", "1m"), false, /**
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 
			 * 是否使用基数排序
			 */
			false/** 不排序 */
			);
		}
	}

	@Override
	public List<RowDataPacket> getRowDataPacketList(byte[] eof) {
		return null;
	}

	private UnsafeRow unsafeRow = null;
	private BufferHolder bufferHolder = null;
	private UnsafeRowWriter unsafeRowWriter = null;
	private int Index = 0;

	@Override
	public void run() {

		if (!running.compareAndSet(false, true)) {
			return;
		}

		boolean nulpack = false;

		try {
			for (;;) {
				final PackWraper pack = packWraperQueue.poll();

				if (pack == null) {
					nulpack = true;
					break;
				}
				if (pack == END_FLAG_PACK) {
					/**
					 * 最后一个节点datenode发送了row eof packet说明了整个
					 * 分片数据全部接收完成，进而将结果集全部发给你 客户端
					 */
					final int warningCount = 0;
					final EofPacket eofp = new EofPacket();
					final ByteBuffer eof = ByteBuffer.allocate(9);
					ByteBufferUtil.writeInt3(eof, eofp.computePacketSize());
					eof.put(eofp.packetId);
					eof.put(eofp.fieldCount);
					ByteBufferUtil.writeInt2(eof, warningCount);
					ByteBufferUtil.writeInt2(eof, eofp.status);
					final SessionContext source = multiQueryHandler.getSession().getSessionContext();
					final byte[] array = eof.array();

					Iterator<UnsafeRow> iters = null;

					if (unsafeRowGrouper != null) {
						/**
						 * group by里面需要排序情况
						 */
						if (globalSorter != null) {
							iters = unsafeRowGrouper.getResult(globalSorter);
						} else {
							iters = unsafeRowGrouper.getResult(globalMergeResult);
						}

					} else if (globalSorter != null) {

						iters = globalSorter.sort();

					} else if (!isStreamOutputResult) {

						iters = globalMergeResult.sort();

					}

					if (iters != null)
						multiQueryHandler.outputMergeResult(source, array, iters);

					if (unsafeRowGrouper != null) {
						unsafeRowGrouper.free();
						unsafeRowGrouper = null;
					}

					if (globalSorter != null) {
						globalSorter.cleanupResources();
						globalSorter = null;
					}

					if (globalMergeResult != null) {
						globalMergeResult.cleanupResources();
						globalMergeResult = null;
					}

					break;
				}

				unsafeRow = new UnsafeRow(fieldCount);
				bufferHolder = new BufferHolder(unsafeRow, 0);
				unsafeRowWriter = new UnsafeRowWriter(bufferHolder, fieldCount);
				bufferHolder.reset();

				/**
				 * 构造一行row，将对应的col填充.
				 */
				BytesHolder mm = new BytesHolder(pack.rowData);
				mm.readInt3();
				mm.read();

				for (int i = 0; i < fieldCount; i++) {
					byte[] colValue = mm.readBytesWithLength();
					if (colValue != null)
						unsafeRowWriter.write(i, colValue);
					else
						unsafeRow.setNullAt(i);
				}

				unsafeRow.setTotalSize(bufferHolder.totalSize());

				if (unsafeRowGrouper != null) {
					unsafeRowGrouper.addRow(unsafeRow);
				} else if (globalSorter != null) {
					globalSorter.insertRow(unsafeRow);
				} else {
					globalMergeResult.insertRow(unsafeRow);
				}

				unsafeRow = null;
				bufferHolder = null;
				unsafeRowWriter = null;
			}

		} catch (final Exception e) {
			multiQueryHandler.handleDataProcessException(e);
		} finally {
			running.set(false);
			if (nulpack && !packWraperQueue.isEmpty()) {
				this.run();
			}
		}
	}

	/**
	 * 释放DataNodeMergeManager所申请的资源
	 */
	public void clear() {

		unsafeRows.clear();

		if (unsafeRowGrouper != null) {
			unsafeRowGrouper.free();
			unsafeRowGrouper = null;
		}

		if (globalSorter != null) {
			globalSorter.cleanupResources();
			globalSorter = null;
		}

		if (globalMergeResult != null) {
			globalMergeResult.cleanupResources();
			globalMergeResult = null;
		}
	}
}
